#pragma once

#include <iostream>
#include <vector>
#include <cassert>
#include <string.h>
#include <stdarg.h>
#include <ctime>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <functional>
#include <sys/types.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unordered_map>
#include <sys/eventfd.h>
#include <mutex>
#include <thread>
#include <sys/timerfd.h>
#include <any>
#include <condition_variable>

#define BUFFERDEFAULTSIZE 1024
#define MAX_LISTEN 20
#define INF 0
#define DBG 1
#define ERR 2

#define LOG_LEVEL DBG

#define LOG(level, format, ...)                                                             \
    do                                                                                      \
    {                                                                                       \
        if (level < LOG_LEVEL)                                                              \
            break;                                                                          \
        time_t lt = time(NULL);                                                             \
        struct tm *ltm = localtime(&lt);                                                    \
        char ti[32] = {0};                                                                  \
        strftime(ti, 31, "%H:%M:%S", ltm);                                                  \
        fprintf(stdout, "[%s %s %d] " #format "\n", ti, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

class Buffer
{
private:
    std::vector<char> _buffer;
    uint64_t _reader_idx;
    uint64_t _writer_idx;

public:
    Buffer()
        : _buffer(BUFFERDEFAULTSIZE),
          _reader_idx(0),
          _writer_idx(0)
    {
    }

    char *Begin()
    {
        return &(*_buffer.begin());
    }

    /*获取当前的写入起始地址*/
    char *WritePosition()
    {
        return Begin() + _writer_idx;
    }

    /*获取当前的读取起始地址*/
    char *ReadPosition()
    {
        return Begin() + _reader_idx;
    }

    /*获取缓冲区末尾空闲空间大小 --也就是写偏移之后的剩余空间*/
    uint64_t TailIdleSize()
    {
        return _buffer.size() - _writer_idx;
    }

    /*获取缓冲区起始空闲空间大小 --也就是读偏移之前的空闲空间*/
    uint64_t HeadIdleSize()
    {
        return _reader_idx;
    }

    /*获取可读数据大小*/
    uint64_t ReadAbleSize()
    {
        return _writer_idx - _reader_idx;
    }

    /*将读偏移向后移动*/
    void MoveReadOffset(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        _reader_idx += len;
    }

    /*将写偏移向后移动*/
    void MoveWriteOffset(uint64_t len)
    {
        assert(len <= TailIdleSize());
        _writer_idx += len;
    }

    /*确保可写空间足够（整体的空间够了就移动数据， 否则扩容)*/
    void EnsureWriteSpace(uint64_t len)
    {
        /*如果末尾空间大小足够，直接返回*/
        if (TailIdleSize() >= len)
        {
            return;
        }
        /*如果末尾空间不够，则判断加上起始位置的空闲空间大小是否足够*/
        if (len <= TailIdleSize() + HeadIdleSize())
        {
            /*将数据移动到起始位置*/
            uint64_t rsz = ReadAbleSize();                            /*把当前数据大小先保存起来*/
            std::copy(ReadPosition(), ReadPosition() + rsz, Begin()); /*把可读数据拷贝到起始位置*/
            _reader_idx = 0;
            _writer_idx = rsz;
        }
        else
        {
            _buffer.resize(_writer_idx + len);
        }
    }

    /*写入数据*/
    void Write(void *data, uint64_t len)
    {
        /*保证有足够空间*/
        EnsureWriteSpace(len);
        /*拷贝数据进去*/
        const char *d = (const char *)data;
        std::copy(d, d + len, WritePosition());
    }
    void WriteAndPush(void *data, uint64_t len)
    {
        Write(data, len);
        MoveWriteOffset(len);
    }

    void WriteString(const std::string &data)
    {
        return Write((void *)data.c_str(), data.size());
    }

    void WriteStringAndPush(const std::string &data)
    {
        WriteString(data);
        MoveWriteOffset(data.size());
    }

    void WriteBuffer(Buffer &data)
    {
        return Write(data.ReadPosition(), data.ReadAbleSize());
    }

    void BufferAndPush(Buffer &data)
    {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize());
    }

    /*读取数据*/
    void Read(void *buf, uint64_t len)
    {
        /*要求获取的数据大小必须小于可读数据大小*/
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(), ReadPosition() + len, (char *)buf);
    }

    void ReadAndPop(void *buf, uint64_t len)
    {
        Read(buf, len);
        MoveReadOffset(len);
    }

    std::string ReadAsString(uint64_t len)
    {
        assert(len <= ReadAbleSize());

        std::string str;
        str.resize(len);
        Read(&str[0], str.size());
        return str;
    }

    std::string ReadAsStringAndPop(uint64_t len)
    {
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }

    char *FindCRLF()
    {
        void *res = memchr(ReadPosition(), '\n', ReadAbleSize());
        return (char *)res;
    }

    /*通常获取一行数据， 这种情况针对是*/
    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == nullptr)
            return "";
        return ReadAsString(pos - (char *)ReadPosition() + 1);
    }

    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }

    /*清空缓冲区*/
    void Clear()
    {
        /*只需要将偏移量归零即可 */
        _reader_idx = 0;
        _writer_idx = 0;
    }
};

class Socket
{
private:
    int _sockfd;

public:
    Socket()
        : _sockfd(-1)
    {
    }
    Socket(int sockfd)
        : _sockfd(sockfd)
    {
    }
    ~Socket()
    {
        Close();
    }

    int Fd()
    {
        return _sockfd;
    }

    /*创建套接字*/
    bool Creat()
    {
        // int socket(int addin, int sockaddr, int PROT)
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ERR_LOG("CREAT SOCKET ERROR");
            return false;
        }
        return true;
    }

    /*绑定地址信息*/
    bool Bind(uint16_t port, const std::string ip)
    {
        // int bind(int sockfd, int sockaddr, int socklen)
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);

        int ret = bind(_sockfd, (struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("BIND ERROR");
            return false;
        }
        return true;
    }

    /*开始监听*/
    bool Listen(int backlog = MAX_LISTEN)
    {
        // int listen(int sockfd, int backlog);
        int ret = listen(_sockfd, backlog);
        if (ret < 0)
        {
            ERR_LOG("LISTEN ERROR");
            return false;
        }
        return true;
    }

    /*发起新连接*/
    bool Connection(uint16_t port, const std::string ip)
    {
        // int connect(int sockfd, struct sockaddr* addr, socklen_t len);
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = connect(_sockfd, (struct sockaddr *)&addr, len); /*向服务端发起连接。 发起连接就默认是listensock接收*/
        if (ret < 0)
        {
            ERR_LOG("CONNECT ERROR");
            return false;
        } //
        return true;
    }

    /*接收新连接*/
    int Accept()
    {
        // int accept(int sockfd, struct sockaddr* addr, socklen_t len);
        int ret = accept(_sockfd, NULL, NULL); /*返回的新的文件描述符， 这个文件描述符*/
        if (ret < 0)
        {
            ERR_LOG("ACCEPT ERROR");
            return -1;
        }
        return ret;
    }

    /*接收数据*/
    size_t Read(void *buf, size_t len, int flag = 0)
    {
        if (len == 0)
            return 0;
        // int recv(int sockfd, void* buf, size_t len, int flags);
        int ret = recv(_sockfd, buf, len, flag); /*从文件中读取数据到缓冲区*/
        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ERR_LOG("RECV ERROR");
            return -1;
        }
        return ret;
    }

    size_t NonBlockRead(void *buf, size_t len)
    {
        return Read(buf, len, MSG_DONTWAIT);
    }

    /*发送数据*/
    size_t Send(void *buf, size_t len, int flag = 0)
    {
        // int send(int sockfd, void* buf, size_t len, int flag);
        int ret = send(_sockfd, buf, len, flag);
        if (ret < 0)
        {
            ERR_LOG("SEND ERROR");
            return ret;
        }
        return ret;
    }

    size_t NonBlockSend(void *buf, size_t len)
    {
        return Send(buf, len, MSG_DONTWAIT);
    }

    /*关闭套接字*/
    void Close()
    {
        if (_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }

    /*创建一个新的服务端链接*/
    bool CreateServer(uint16_t port, std::string ip = "0.0.0.0", int block_flag = 0)
    {
        // 创建套接字， 绑定地址信息， 监听套接字， 设置地址重用
        if (!Creat())
            return false;
        if (block_flag)
            SetNonblock();
        else
            SetBlock();
        if (!Bind(port, ip))
            return false;
        if (!Listen())
            return false;
        Reuse();
        return true;
    }

    /*创建一个新的客户端链接*/
    bool CreateClient(uint16_t port, std::string ip)
    {
        // 客户端也要有自己的套接字
        if (!Creat())
            return false;
        // 链接远端
        if (!Connection(port, ip))
            return false;
        return true;
    }

    /*设置地址重用*/
    void Reuse()
    {
        // 设置地址重用， int setsockopt(int sockfd, int level, int opename, const void* optval, socklen_t optlen);
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void *)&val, sizeof(int));
    }

    /*设置非阻塞*/
    bool SetNonblock()
    {
        // int fcntl(int sockfd, int cmd, int flag);

        int fl = fcntl(_sockfd, F_GETFL, 0);
        int ret = fcntl(_sockfd, F_SETFL, fl | O_NONBLOCK);
        if (ret < 0)
        {
            ERR_LOG("SETNONBLOCK ERROR");
            return false;
        }
        return true;
    }
    bool SetBlock()
    {
        // int fcntl(int sockfd, int cmd, int flag);

        int fl = fcntl(_sockfd, F_GETFL, 0);
        int ret = fcntl(_sockfd, F_SETFL, fl & ~O_NONBLOCK);
        if (ret < 0)
        {
            ERR_LOG("SETNONBLOCK ERROR");
            return false;
        }
        return true;
    }
};

class Poller;
class EventLoop;
class TimeWheel;

class Channel
{
private:
    EventLoop *_loop; /*连接所关联的一个loop*/
    int _fd;
    uint32_t _events;  // 当前要监控的事件
    uint32_t _revents; // 当前连接实际触发的事件
    using EventCallback = std::function<void()>;
    EventCallback _reader_callback;
    EventCallback _writer_callback;
    EventCallback _error_callback;
    EventCallback _close_callback;
    EventCallback _event_callback;

public:
    Channel(EventLoop *loop, int fd)
        : _fd(fd),
          _loop(loop),
          _events(0),
          _revents(0)
    {
    }
    int Fd() { return _fd; }
    void SetReadCallback(const EventCallback &cb) { _reader_callback = cb; }
    void SetWriteCallback(const EventCallback &cb) { _writer_callback = cb; }
    void SetErrorCallback(const EventCallback &cb) { _error_callback = cb; }
    void SetCloseCallback(const EventCallback &cb) { _close_callback = cb; }
    void SetEventCallback(const EventCallback &cb) { _event_callback = cb; }
    uint32_t GetEvents()
    {
        return _events;
    }
    void SetEvents(uint32_t event) /*就绪实际的事件*/
    {
        _revents = event;
    }
    bool ReadAble() { return (_events & EPOLLIN); }   // 当前是否可读
    bool WriteAble() { return (_events & EPOLLOUT); } // 当前是否可写
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Updata();
    } // 启动读事件监控
    void EnableRead()
    {
        _events |= EPOLLIN;
        Updata();
    } // 启动写事件监控
    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Updata();
    } // 关闭读事件监控
    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Updata();
    } // 关闭写事件监控
    void DisableAll()
    {
        _events = 0;
        Updata();
    } // 关闭所有时间监控
    void Remove(); // 移除监控
    void Updata(); // 添加监控
    void HandleEvent()
    {
        if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if (_reader_callback)
                _reader_callback();
            /*不管什么时候, 都调用的回调函数*/
            if (_event_callback)
                _event_callback();
        }
        /*有可能会释放连接的操作事件， 一次只处理一个*/
        if (_revents & EPOLLOUT)
        {
            if (_writer_callback)
                _writer_callback();

            /*放到事件处理完之后调用， 刷新活跃度*/
            if (_event_callback) /*只有当数据发过去之后， 才刷新活跃度。 */
                _event_callback();
        }
        else if (_revents & EPOLLERR)
        {
            /*一旦出错， 就会释放连接， 因此要放到前边任意回调*/
            if (_event_callback)
                _event_callback();
            if (_error_callback)
                _error_callback();
        }
        else if (_revents & EPOLLHUP)
        {
            /*不管什么时候, 都调用的回调函数*/
            if (_event_callback)
                _event_callback();
            if (_close_callback)
                _close_callback();
        }

    } // 事件处理，一旦链接出发了事件， 就调用这个函数
};

#define MAX_EPOLLEVENTS 1024

class Poller
{
private:
    int _epfd;
    struct epoll_event events[MAX_EPOLLEVENTS]; // 这里面保存的是：某个fd实际触发了某个事件
    std::unordered_map<int, Channel *> _chn_hash;

private:
    /*对epoll的直接操作*/
    void Updata(Channel *chn, int op)
    {
        // Updata就是使用epoll_ctl对事件进行控制
        // int epoll_ctl(int epfd, int op, int fd, epoll_event* event);
        int fd = chn->Fd();
        epoll_event ev;
        ev.data.fd = fd;
        ev.events = chn->GetEvents();
        int ret = epoll_ctl(_epfd, op, fd, &ev); // 删除某个文件描述符的某个事件
        if (ret < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("Epoll控制错误, 错误信息为: %s\n", strerror(errno));
        }
    }
    /*判断一个channel是否已经添加了事件监控*/
    bool HashChannel(Channel *chn)
    {
        // 判断一个channel里面是否添加了事件监控，只需要遍历整个channels, 看看里面是否存在chn即可
        auto it = _chn_hash.find(chn->Fd());
        if (it == _chn_hash.end()) /*如果没找到， 就返回false*/
        {
            return false;
        }
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(10000);
        if (_epfd < 0)
        {
            ERR_LOG("创建epoll模型错误");
        }
    }
    /*更新或添加监控*/
    void EventUpdata(Channel *chn)
    {
        /*先查看存不存在该事件*/
        bool ret = HashChannel(chn);
        if (!ret) /*没找到， 就要进入这恶鬼里面添加fd的监控*/
        {
            /*不存在这个事件， 不存在就添加, 添加到哪里？ 添加到epoll模型的底层红黑树中*/
            Updata(chn, EPOLL_CTL_ADD);
            _chn_hash.insert({chn->Fd(), chn});
        }
        /*存在就修改*/
        Updata(chn, EPOLL_CTL_MOD);
    }
    /*移除监控*/
    void MoveEvent(Channel *chn)
    {
        auto it = _chn_hash.find(chn->Fd());
        if (it != _chn_hash.end())
        {
            _chn_hash.erase(it);
        }
        Updata(chn, EPOLL_CTL_DEL);
    }
    /*拿到就绪文件描述符和事件*/
    void Poll(std::vector<Channel *> *active)
    {
        //
        int ret = epoll_wait(_epfd, events, MAX_EPOLLEVENTS, -1);   /*这里不应该是阻塞吗？因为滴答*/
        if (ret < 0)
        {
            if (errno == EINTR)
            {
                ERR_LOG("信号中断， 原因：%s", strerror(errno));
                return;
            }
            abort();
        }
        else if (ret == 0)
        {
            DBG_LOG("没有新的事件就绪");
        }
        else if (ret > 0)
        {
            // 如果存在
            for (int i = 0; i < ret; i++)
            {
                // 遍历整个events， 然后拿到对应元素的事件管理器
                auto it = _chn_hash.find(events[i].data.fd);
                assert(it != _chn_hash.end());           /*判断it存在不存在*/
                it->second->SetEvents(events[i].events); /*这个就是给fd更新实际就绪的事件 */
                active->push_back(it->second);
            }
        }
    }
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;

class TimeTask
{
private:
    /*首先要有定时任务的id，
    然后要有定时任务的超时时间
    标记定时任务是否被取消
    还有定时任务本身要执行的操作
    最后还有一个用于删除TimeWheel中保存
    的定时器对象的操作*/

    uint64_t _task_id; /*定时器id是一个8字节整形*/
    uint32_t _timeout; /*超时时间是4字节*/
    bool _cancel;
    TaskFunc _task_func;
    ReleaseFunc _release_func;

public:
    /*然后定时器的操作， 首先构造
    然后设置释放定时任务的函数
    然后标记取消位置
    获取超时时间
    析构
    */
    TimeTask(uint64_t id, uint32_t timeout, TaskFunc cb)
        : _task_id(id),
          _timeout(timeout),
          _cancel(0),
          _task_func(cb),
          _release_func(nullptr)
    {
    }

    void SetRelease(const ReleaseFunc &cb)
    {
        _release_func = cb;
    }

    void Cancel() // 这个没有说明没有了
    {
        _cancel = true; // 这里的this指针是空。 说明
    }

    uint32_t GetDelay()
    {
        return _timeout;
    }
    ~TimeTask() /*什么情况下调用这个函数， 只有当时间轮中最后一个定时器指针对象被干掉了， 才会执行这个函数*/
    {
        DBG_LOG("yes, i has release, %p", this);
        if (_cancel == false)
            _task_func();
        _release_func();
    }
};

#define WHEELMAXSIZE 1024
class TimeWheel
{
private:
    /*首先有表针， 也就是循环队列的下标。
    然后有表盘最大容量， 也就是循环队列的最大容量。
    然后就是表盘， 就是循环队列。
    然后就是表盘中已经存储的定时任务。*/
    using Wptr = std::weak_ptr<TimeTask>;
    using Sptr = std::shared_ptr<TimeTask>;
    int _tick;     /*表针*/
    int _capacity; /*表盘最大容量*/
    std::vector<std::vector<Sptr>> _wheel;
    std::unordered_map<uint64_t, Wptr> _timers;

    EventLoop *_loop;
    int _timerfd;            /*定时器描述符*/
    Channel *_timer_channel; /*定时器的事件监控对象， 用于在_loop中进行可读事件回调*/
private:
    void TimeRemove(uint64_t id)
    {
        auto it = _timers.find(id); /*就是寻找定时器id， 然后找到删除就行了。*/
        if (it != _timers.end())
        {
            _timers.erase(it);
            return;
        }
    }

    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ERR_LOG("TIMERFD CREATE FAILED");
            abort();
        }

        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0; //*第一次超时时间为1秒
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0; // 第一次超时后， 每次超时的间隔
        timerfd_settime(timerfd, 0, &itime, NULL);

        return timerfd;
    }

    void ReadTimefd()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, sizeof(times));
        if (ret < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return;
            }
            ERR_LOG("READ ERROR");
            abort();
        }
        return;
    }
    void OneStep()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear();
    }

    void OnTime()
    {
        ReadTimefd();
        OneStep();
    }

    void TimeaddInLoop(uint64_t id, uint32_t timeout, TaskFunc func)
    {
        Sptr pt(new TimeTask(id, timeout, func));
        /*设定pt里面的销毁函数*/
        pt->SetRelease(std::bind(&TimeWheel::TimeRemove, this, id)); /*release时时间轮里面的TimerRemove函数*/
        _timers[id] = Wptr(pt);                                      /*添加到哈希表中*/

        int pos = (_tick + timeout) % _capacity;
        _wheel[pos].push_back(pt);
    }

    void CancelInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end())
        {

            auto pt = (it->second).lock(); /*问题是， wptr返回了空，wptr里面保存的是什么？
                                              是定时器。 这个定时器什么时候被释放呢？就是在滴答的时候
                                              ，要先析构调用完， 定时器才被释放。现在没调用完， 为什么被释放了？
                                              sptr引用计数提前为0了？但是提前为0也是调用析构， 也就是sptr引用
                                              计数为0和调用析构是绑定的。 那为什么？*/

            if (pt != nullptr)
                pt->Cancel(); //
        }
    }
    void TimefreshInLoop(uint64_t id)
    {

        auto it = _timers.find(id);
        if (it != _timers.end())
        {
            Sptr pt = _timers[id].lock();
            int pos = (_tick + pt->GetDelay()) % _capacity;
            _wheel[pos].push_back(pt); /*刷新是将是将论里面的数据拷贝一份*/
        }
    }

public:
    TimeWheel(EventLoop *loop, int capacity = 60, int tick = 0)
        : _capacity(capacity),
          _tick(tick),
          _wheel(_capacity),
          _loop(loop),
          _timerfd(TimeWheel::CreateTimerfd()),
          _timer_channel(new Channel(_loop, _timerfd))
    {
        _timer_channel->SetReadCallback(std::bind(&TimeWheel::OnTime, this));
        _timer_channel->EnableRead(); /*启动可读监控*/
    }

    void TimerAdd(uint64_t id, uint32_t timeout, TaskFunc func); // 在这里添加的。 这个添加函数是利用上层的func添加到

    void TimerCancel(uint64_t id);

    void TimerFresh(uint64_t id);

    //*这个接口存在线程安全问题， 这个接口实际上不能被外界使用者调用， 只能在模块内部， 在对应的EventLoop线程内执行
    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
            return false;
        return true;
    }
};

class EventLoop
{
private:
    using Functor = std::function<void()>;
    Poller _poller; /*管理epoll模型*/
    int _event_fd;  /*防止监控事件的时候阻塞*/
    Channel _event_channel;
    std::vector<Functor> _task; /*任务池*/
    std::mutex _mutex;          /*锁*/
    std::thread::id _thread_id; /*eventloop的线程id*/
    TimeWheel _timer_wheel;

public:
    /*执行所有的任务*/
    void RunAllTask()
    {
        // 先创建一个数组， 将task里面的所有任务都拿到这个数组中。同时清空原数组
        std::vector<Functor> functors;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _task.swap(functors);
        }

        // 执行所有的任务
        for (auto &e : functors)
        {
            e();
        }
    }

    static int Createventfd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            ERR_LOG("CREAT EVENTFD FAILED");
        }
        return efd;
    }
    void ReadEventfd()
    {
        uint64_t res;
        int ret = read(_event_fd, &res, sizeof(res));
        if (ret < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return;
            }
            ERR_LOG("EVENTFD READ FAILED");
            return;
        }
    }
    void WeakUpEventfd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(val));
        if (ret < 0)
        {
            if (errno == EINTR)
            {
                return;
            }

            ERR_LOG("EVENTFD WRITE FAILED");
            abort();
        }
        return;
    }

public:
    /*构造函数*/
    EventLoop()
        : _thread_id(std::this_thread::get_id()),
          _event_fd(EventLoop::Createventfd()),
          _event_channel(this, _event_fd),
          _timer_wheel(this)
    {
        _event_channel.SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
        _event_channel.EnableRead();
    }

    /*添加或者修改文件描述符的监控*/
    void UpdataEvent(Channel *_chn)
    {
        _poller.EventUpdata(_chn);
    }

    /*删除文件描述符的监控*/
    void RemoveEvent(Channel *_chn)
    {
        _poller.MoveEvent(_chn);
    }

    /*判断是否和eventloop是同一个线程*/
    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }

    bool AssertInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }

    /*判断当前线程和eventloop是否是同一个线程， 如果是同一个线程，
     那么就直接执行， 如果不是同一个线程， 就把操作压入到任务池*/
    void RunInLoop(const Functor &cb)
    {
        if (IsInLoop())
        {
            cb();
            return;
        }
        QueueInLoop(cb);
    }

    /*将任务压入到任务池当中*/
    void QueueInLoop(const Functor &cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _task.push_back(cb); //*这里是将所有的任务压倒任务池中，去任务池中执行定时器任务
        }

        /*压入到任务池之后， 就唤醒eventfd*/
        WeakUpEventfd();
    }

    /*添加一个定时任务*/
    void AddTimer(uint64_t id, uint32_t delay, const TaskFunc &cb)
    {
        _timer_wheel.TimerAdd(id, delay, cb);
    }

    void TimerRefresh(uint64_t id)
    {
        _timer_wheel.TimerFresh(id);
    }

    void TimerCancel(uint64_t id)
    {
        _timer_wheel.TimerCancel(id);
    }

    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.HasTimer(id);
    }

    /*执行， 三步走：时间监控——》就绪事件处理——》执行任务*/
    void Start()
    {
        while (1)
        {
            std::vector<Channel *> actives;
            _poller.Poll(&actives); /*_poller进行监控*/
            for (auto e : actives)
            {
                e->HandleEvent();
            }
            // 执行所有任务
            RunAllTask();
        }
    }
};

void Channel::Updata()
{
    _loop->UpdataEvent(this);
}
void Channel::Remove()
{
    _loop->RemoveEvent(this);
}

void TimeWheel::TimerAdd(uint64_t id, uint32_t timeout, TaskFunc func) // 这个添加函数是loop里面的RunInLoop添加定时器。对吗？ 定时器使一个Sptr指针，以weakptr的形象保存到了timers哈希表中。 然后又以人物的形式每隔几秒保存到了时间轮中。 他的本质是一个TimeTask。这里面有释放函数， 又执行函数， 还有cancel标志。 还有id。
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimeaddInLoop, this, id, timeout, func));
}

void TimeWheel::TimerCancel(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::CancelInLoop, this, id));
}

void TimeWheel::TimerFresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimefreshInLoop, this, id));
}

typedef enum
{
    DISCONNECTED, /*连接关闭状态*/
    CONNECTING,   /*连接建立好后的各项设置的设置中状态*/
    CONNECTED,    /*各种设置已完成， 正常通信的状态*/
    DISCONNECTING /*待关闭状态*/
} ConnState;
/*CONNECTED -- 连接建立完成， 各种设置已经完成， 可以通信的状态。
DISCONNECTING -- 待关闭状态*/
class Connection;
using PtrConn = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    EventLoop *_loop;  /*连接所关联的一个EventLoop。*/
    uint64_t _conn_id; /*连接的唯一的ID， 便于连接的管理和查找*/
    // uint64_t _timerfd;   /*定时器ID， 必须是唯一的，这里为了简化， 所以使用_conn_id作为了定时器ID*/
    int _sockfd;                   /*连接关联的文件描 述符*/
    bool _enable_inactive_release; /*连接是否启动非活跃销毁的判断标志， 默认是false*/
    ConnState _state;
    Socket _socket;     /*套接字管理*/
    Channel _channel;   /*连接的事件管理*/
    Buffer _in_buffer;  /*输入缓冲区 -- 存放从socket中读取到的数据*/
    Buffer _out_buffer; /*输出缓冲区 -- 存放要发给对端的数据*/
    std::any _context;  /*请求的接受处理上下文*/

    // 这四个回调函数， 是让服务器模块来设置的， 其实服务器模块的处理回调也是组件使用者设置的
    // 换句话说，这几个都是组件使用者使用的
    using ConnectedCallback = std::function<void(const PtrConn &)>;
    using MessageCallback = std::function<void(const PtrConn &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConn &)>;
    using AnyEventCallback = std::function<void(const PtrConn &)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _close_callback;
    AnyEventCallback _anyevent_callback;

    /*组件内的连接关闭回调 -- 组件内设置的。因为服务器组件内会把所有的连接管理起来。
    一旦某个连接要关闭，就应该从管理的地方移除掉自己的信息。*/
    ClosedCallback _server_close_callback;

private:
    /*五个channel的事件回调函数*/
    void HandleRead() // 描述符出发可读事件后调用的函数， 接收socket数据放到接收缓冲区中，然后调用_message_callback进行业务处理
    {
        char buffer[65536];
        int ret = _socket.NonBlockRead(buffer, sizeof(buffer)); /*讲述读取到用户缓冲区中，然后将数据写到inbuffer里面。 如果里面大于零
      就进行处理*/
        if (ret >= 0)
        {
            buffer[ret] = 0;
            _in_buffer.WriteAndPush(buffer, ret);
            if (_in_buffer.ReadAbleSize() > 0)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        else
        {
            return ShutDownInLoop();
        }
    }
    void HandleWrite() // 描述符可写事件触发后调用的函数，将发送缓冲区中的数据进行发送
    {
        /*发送缓冲区的可读区域就是要发送的数据*/
        int ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        if (ret < 0)
        {
            /*写入失败， 这个时候判断一下写入缓冲区是否有数据， 有数据就处理。 然后真正的关闭连接*/
            if (_in_buffer.ReadAbleSize() > 0)
            {
                _message_callback(shared_from_this(), &_in_buffer);
            }
            ReleaseInLoop();

            return;
        }
        else if (ret == 0)
        {
            _channel.DisableWrite();
        }
        _out_buffer.MoveReadOffset(ret);
        /*如果没有写入失败，则判断一下可写区域是否还有剩余， 如果满了， 那么关闭可写。
        如果是待关闭状态， 直接关闭 */
        if (_out_buffer.ReadAbleSize() == 0)
        {
            _channel.DisableWrite();
        }
        if (_state == DISCONNECTING)
        {
            ReleaseInLoop();
        }
    }

    void HandleClose() // 描述符触发挂断时间后调用的函数
    {
        if (_in_buffer.ReadAbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        ReleaseInLoop();
    }
    void HandleError() // 描述符触发出错事件
    {
        HandleClose();
    }
    void HandleEvent() // 描述符触发任意事件
    {
        if (_enable_inactive_release == true)
        {
            _loop->TimerRefresh(_conn_id); /*如果这个时候的服务器的非活跃来接启动了， 那么就去刷新他  */
        }
        if (_anyevent_callback)
        {
            _anyevent_callback(shared_from_this());
        }
    }

    void EstablishedInLoop() /*连接获取之后， 所处的状态下，c
                                  要进行各种设置（给channel进行事件回调，
                                  以及启动读监控）*/
    {
        // 修改连接状态    //设置时间监控       //调用回调函数
        assert(_state == CONNECTING);
        _state = CONNECTED;
        _channel.EnableRead(); /*可读， 对方发过来数据， 连接就可以读取到数据。 */
        if (_anyevent_callback)
            _anyevent_callback(shared_from_this());
    }
    void ReleaseInLoop() /*这个接口， 才是实际的释放接口*/
    {
        /*修改连接状态， 将其设置为DISCONNECTED*/
        _state = DISCONNECTED;

        /*移除连接的事件监控*/
        _channel.Remove();

        /*关闭描述符*/
        _socket.Close();

        /*如果当前定时器队列中还有定时器任务，则取消任务, 这个就是将TimeTask里面的Cancel置为true*/
        if (_loop->HasTimer(_conn_id))
            CancelInactiveReleaseInLoop(); // 问题是导致了时间轮里面的sharedptr没了， 但是
                                           // timers里面的weak_Ptr还存在
        /*调用关闭回调函数，避免移除服务器管理的连接信息导致connection被释放，再去处理会出错，
        因此先调用用户的回调函数*/
        if (_close_callback)
            _close_callback(shared_from_this());

        /*调用服务器的释放函数*/
        _server_close_callback(shared_from_this());
    }
    void SendInLoop(char *data, size_t len)
    {
        if (_state == DISCONNECTED)
            return;                          /*如果处于已关闭状态， 直接退出*/
        _out_buffer.WriteAndPush(data, len); /*写入到out缓冲区中。*/
        if (_channel.WriteAble() == false)   /*如果连接中不可写，那么就变成可写*/
        {
            _channel.EnableWrite();
        }
    }
    /*这个关闭并非实际的连接释放操作*/
    void ShutDownInLoop()
    {
        _state = DISCONNECTING; /*半关闭状态*/
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        if (_out_buffer.ReadAbleSize() > 0)
        {
            if (_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }
        if (_out_buffer.ReadAbleSize() == 0)
        {
            ReleaseInLoop();
        }
    }

    /*启动非活跃连接释放规则*/
    void EnableInactiveReleaseInLoop(int sec) /*非活跃连接销毁的逻辑就是将非活跃标志置为1.然后添加销毁任务*/
    {
        // 将判断标志，_enable_inactive_release置为true。
        _enable_inactive_release = true;
        // 添加定时销毁任务。 如果已经存在， 延迟一下即可
        if (_loop->HasTimer(_conn_id))
        {
            return _loop->TimerRefresh(_conn_id);
        }
        // 如果不存在，则新增
        _loop->AddTimer(_conn_id, sec, std::bind(&Connection::ReleaseInLoop, this));
    }
    void CancelInactiveReleaseInLoop()
    {
        printf("调用到了CancelInactiveReleaseInLoop\n");
        _enable_inactive_release = false;
        _loop->TimerCancel(_conn_id);
    }
    void UpgradeInLoop(const std::any &context, const ConnectedCallback &conn,
                       const MessageCallback &msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _context = context;
        _message_callback = msg;
        _connected_callback = conn;
        _close_callback = closed;
        _anyevent_callback = event;
    }

public:
    Connection(EventLoop *loop, uint64_t id, int sockfd)
        : _channel(loop, _sockfd),
          _conn_id(id),
          _sockfd(sockfd),
          _enable_inactive_release(false),
          _loop(loop),
          _state(CONNECTING),
          _socket(_sockfd)
    {
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));

        /*启动读事件操作不能在构造函数中完成*/
    }

    ~Connection()
    {
        DBG_LOG("release connection address: %p", this);
    }

    int Fd()
    {
        return _sockfd;
    }
    int Id()
    {
        return _conn_id;
    }
    ConnState state()
    {
        return _state;
    } /*返回状态*/
    bool Connected() /*是否处于CONNECTED状态*/
    {
        return _state == CONNECTED;
    }
    void SetContext(const std::any &context) // 设置上下文，
    {
        _context = context;
    }
    std::any *GetContext() // 获取上下文
    {
        return &_context;
    }
    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb)
    {
        _close_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _anyevent_callback = cb;
    }
    void SetServerClosedCallback(const ClosedCallback &cb)
    {
        _server_close_callback = cb;
    }

    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    void Send(char *data, size_t len)
    {
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, data, len));
    }
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    void EnableInactiveRelease(int sec) /*设置非活跃连接销毁就是让loop去执行非活跃连接*/
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    /*这个接口必须在EventLoop中立即执行，防备新的事件处罚后，处理的时候， 切换任务还没有被执行，会导致数据使用原协议处理*/
    void Upgrade(const std::any &content, const ConnectedCallback &conn,
                 const MessageCallback &msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, content, conn, msg, closed, event));
    }
};

class Acceptor
{
private:
    /*创建套接字连接*/
    Socket _socket;

    /*监听套接字进行事件监控*/
    EventLoop *_loop;

    /*监听套接字进行事件管理*/
    Channel _channel;

    using AcceptCallback = std::function<void(int)>;
    AcceptCallback _accept_callback;

private:
    /*监听套接字的读事件回调处理函数 -- 获取新连接。调用_accept_callback进行新连接处理*/
    void HandleRead()
    {
        int newfd = _socket.Accept();
        if (newfd < 0)
        {
            ERR_LOG("accept error");
            return;
        }
        _accept_callback(newfd);
    }
    int AcreateServer(int port)
    {
        int ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }

public:
    Acceptor(EventLoop *loop, uint16_t port)
        : _loop(loop),
          _socket(AcreateServer(port)),
          _channel(_loop, _socket.Fd())
    {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }

    void SetAcceptCallback(const AcceptCallback &cb)
    {
        _accept_callback = cb;
    }

    void Listen()
    {
        _channel.EnableRead();
    }
};

class LoopThread
{
private:
    std::mutex _mutex;             /*互斥锁*/
    std::condition_variable _cond; /*条件变量*/
    EventLoop *_loop;              /*loop是线程里面的成员， 每一个线程对应一个loo*/
    std::thread _thread;           /*线程*/

private:
    void ThreadEntry()
    {
        EventLoop loop; /*这里不是用new是因为一个loop对应一个线程。 loop和线程应该是绑定起来的*/
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        _loop->Start(); /*执行start。其实就是监听epoll模型里面的文件描述符对应的事件。
       一个loop对应一个epoll模型。 所以每一个线程都会被分配到一些Connection。*/
    }

public:
    LoopThread() /*构造函数就初始化线程， 意味着如果创建LoopThread对象， 那么就直接创建子线程。 执行子线程
 里面的_loop->start()*/
        : _thread(std::bind(&LoopThread::ThreadEntry, this)),
          _loop(nullptr)
    {
    }

    EventLoop *GetLoop()
    {
        EventLoop *loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]()
                       { return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }
};


/*线程数量可配置。
注意事项：在配置里面，线程池数量为0，在服务器中，主从reactor模型是
主线程负责新连接的获取，新县城负责新连接数据处理的事件监控和处理。*/

/*对所有的线程进行管理。
其实就是管理0个或者多个LoopThread对象。*/

/*提供线程分配的功能
当主线程获取了一个新连接， 需要把新连接挂接到从属线程上进行处理。
如果没有从属线程，则直接分配给主线程的eventloop，进行处理。
假如有多个从属线程。 具体给哪一个呢？ 则采用RR轮转分配。进行
线程的分配（将对应线程的Eventloop获取到，设置给对应的Connectione*/


class LoopThreadPool
{
private:
    int _thread_count;
    int _next_index;
    EventLoop *_base_loop;
    std::vector<LoopThread *> _threads;
    std::vector<EventLoop *> _loops;
public:
    LoopThreadPool(EventLoop *loop)
        :_thread_count(0),
        _next_index(0),
        _base_loop(loop)
    {}

    void SetThreadCount(int count)
    {
        _thread_count = count;
    }

    void Create()
    {
        _threads.resize(_thread_count);
        _loops.resize(_thread_count);
        for (int i = 0; i < _thread_count; i++)
        {
            _threads[i] = new LoopThread();
            _loops[i] = _threads[i]->GetLoop();
        }
    }

    EventLoop *Next_loop()
    {
        _next_index = (_next_index + 1) % _thread_count;
        return _loops[_next_index];
    }
};


class TcpServer
{
private:
    uint64_t _conn_id;      /*自动增长的连接id*/
    int _timeout;           /*非活跃连接的统计时间 -- 多长时间无通信就是非活跃链接*/   
    bool _enable_inactive_release;   /*是否启动非活跃链接的判断标志*/
    Acceptor _acceptor;     /*这是监听套接字的管理对象*/
    EventLoop _baseloop;    /*这个是主线程的EventLoop对象，负责监听事件的处理*/
    LoopThreadPool _pool;   /*这是从属线程池， 从属EventLoop的线程池*/
    std::unordered_map<uint64_t, PtrConn> _conns;  /*保存管理所有链接对应的shared_ptr对象*/
    
    
    using ConnectedCallback = std::function<void(const PtrConn &)>;
    using MessageCallback = std::function<void(const PtrConn &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConn &)>;
    using AnyEventCallback = std::function<void(const PtrConn &)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _close_callback;
    AnyEventCallback _anyevent_callback;

private:
    /*为新连接构造一个Connection进行管理*/
    void NewConnection(int fd)
    {

    }
    void RemoveConnection(){} /*管理Connection的_conns中移除连接信息*/


public:

    TcpServer()
    {}

    /*1、先设置线程池的容量*/
    void SetPoolCapacity(int count){}

    /*2345、设置新连接到来的回调函数、设置消息处理的回调函数、关闭回调函数、任意事件处理函数*/
    void SetConnectedCallback(const ConnectedCallback &cb)
    {
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb)
    {
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb)
    {
        _close_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb)
    {
        _anyevent_callback = cb;
    }

    /*6、启动非活跃连接销毁*/
    void EnableInactiveRelease(int timeout){}

    /*7、用于添加一个定时任务*/
    void AfterTask(const Functor &task, int delay){}

    /*8、启动*/
    void Start(){}

};  